/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package edu.isi.pegasus.planner.subworkflow.scheduler.implementation;

import edu.isi.pegasus.planner.classes.ADag;
import edu.isi.pegasus.planner.classes.PegasusBag;
import edu.isi.pegasus.planner.dax.ADAG;
import edu.isi.pegasus.planner.dax.ParADAG;
import edu.isi.pegasus.planner.subworkflow.partitioner.DAXAbstraction;
import edu.isi.pegasus.planner.subworkflow.partitioner.SiteAbstraction;
import edu.isi.pegasus.planner.subworkflow.scheduler.Scheduler;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

/**
 *
 * @author wchen
 */
public abstract class ListScheduler extends Scheduler implements ListSchedulerImplementation{

    private Map weight;
    private Map avail;
    protected Map getChildrenMap;
    protected Map getParentsMap;
    protected DAXAbstraction headNode;
    protected DAXAbstraction rearNode;
    protected Map mSiteList;
    private Map mSitePureSize;

    public ListScheduler(ADag dag, PegasusBag bag)
    {
        super(dag,bag);
        getChildrenMap = new HashMap<String, ArrayList>();
        getParentsMap   = new HashMap<String, ArrayList>();
        avail = new HashMap<SiteAbstraction, Double>();
        weight = new HashMap<DAXAbstraction, Map>();
        mSiteList = new HashMap<SiteAbstraction, ArrayList>();
        mSitePureSize = new HashMap<DAXAbstraction, Double>();
        
        
    }
    protected double getweight(DAXAbstraction bigDax, SiteAbstraction site)
    {

        if(!weight.containsKey(bigDax))
        {
            weight.put(bigDax, new HashMap());
            System.out.println("Warning: this guy doesn't exist:" + bigDax.adag.getName());
            return 0.0;
        }
        Map map = (Map)weight.get(bigDax);

        if(!map.containsKey(site))
        {
            map.put(site, 0.0);
            return 0.0;
        }
        double runtime =(Double)map.get(site);
        return runtime;
    }
    public double getSitePureSize(DAXAbstraction bigDax)
    {
        if(mSitePureSize.containsKey(bigDax))
            return (Double)mSitePureSize.get(bigDax);
        return 0.0;
    }
    protected double getavail(SiteAbstraction site)
    {
        if(!avail.containsKey(site))
            avail.put(site, 0.0);
        return (Double)avail.get(site);
    }
    protected void setavail(SiteAbstraction site, double runtime)
    {
        if(avail!= null)
            avail.put(site, runtime);
        else
            System.err.println("Table Avail doesn't exist");
    }
    protected void getRel()
    {
        for(int i = 0; i < mParentDAX.size(); i++)
        {
            String parent = (String)mParentDAX.get(i);
            String child = (String)mChildDAX.get(i);
            if(!getChildrenMap.containsKey(parent))
                getChildrenMap.put(parent, new ArrayList<String>());
            if(!getParentsMap.containsKey(child))
                getParentsMap.put(child, new ArrayList<String>());
            ArrayList clist = (ArrayList)getChildrenMap.get(parent);
            if(!clist.contains(child))
                clist.add(child);
            ArrayList plist = (ArrayList)getParentsMap.get(child);
            if(!plist.contains(parent))
                plist.add(parent);


        }
    }

        protected void Fake()
    {
        mDaxMap.clear();
        weight.clear();
        getChildrenMap.clear();
        getParentsMap.clear();
        DAXAbstraction dax0 = null, dax1 = null, dax2 = null;
        for(int i = 0; i < 3; i ++)
        {
            ParADAG dag = new ParADAG("subworkflow"+i);
            DAXAbstraction bigDax = new DAXAbstraction(dag);
            if(i==0)
                dax0 = bigDax;
            if(i==1)
                dax1 = bigDax;
            if(i==2)
                dax2 = bigDax;
            mDaxMap.put(bigDax, new HashMap());
             //if(!weight.containsKey(bigDax))
                weight.put(bigDax, new HashMap<SiteAbstraction, Double>());
           getParentsMap.put(bigDax.adag.getName(), new ArrayList());
           getChildrenMap.put(bigDax.adag.getName(), new ArrayList());
           mDAXID2DAX.put(dag.getName(), bigDax);
           Map map = (Map)weight.get(bigDax);

           double runtime = i * 100;
           for(int j = 0; j < mSiteIndex.size(); j++)
                {
                    SiteAbstraction site = (SiteAbstraction)mSiteIndex.get(j);
                    //double runtime = estimator.getEstimatedRuntime(site);
                    //System.out.println("for dax " + bigDax.adag.mName + " runtime: " + runtime );
                    //if(!map.containsKey(site))
                        map.put(site, runtime);
            }

        }
        ArrayList list1 = (ArrayList)getParentsMap.get(dax0.adag.getName());
        list1.add(dax1.adag.getName());
        list1.add(dax2.adag.getName());
        list1 = (ArrayList)getChildrenMap.get(dax1.adag.getName());
        list1.add(dax0.adag.getName());
        list1 = (ArrayList)getChildrenMap.get(dax2.adag.getName());
        list1.add(dax0.adag.getName());



    }

    protected DAXAbstraction getHeadNode()
    {

        headNode = new DAXAbstraction(new ParADAG("HEAD"));
        rearNode = new DAXAbstraction(new ParADAG("REAR"));
        mDAXID2DAX.put(headNode.adag.getName(), headNode);
        mDAXID2DAX.put(rearNode.adag.getName(), rearNode);
        //weight.put(headNode, avail)
        ArrayList headList = new ArrayList<String>();
        getChildrenMap.put(headNode.adag.getName(), headList);

        ArrayList rearList = new ArrayList<String>();
        getParentsMap.put(rearNode.adag.getName(), rearList);
        for (Iterator it = mDaxMap.keySet().iterator(); it.hasNext(); )
        {
            DAXAbstraction bigDax = (DAXAbstraction)it.next();
            if(bigDax.equals(headNode) || bigDax.equals(rearNode))
                continue;

            if(!getParentsMap.containsKey(bigDax.adag.getName()))
                getParentsMap.put(bigDax.adag.getName(), new ArrayList<String>());

            ArrayList list = (ArrayList)getParentsMap.get(bigDax.adag.getName());
            if(list.size() == 0 )
            {
                list.add(headNode.adag.getName());
                headList.add(bigDax.adag.getName());
            }
            if(!getChildrenMap.containsKey(bigDax.adag.getName()))
                getChildrenMap.put(bigDax.adag.getName(), new ArrayList<String>());
            ArrayList list1 = (ArrayList)getChildrenMap.get(bigDax.adag.getName());

            if(list1.size() == 0 )
            {
                list1.add(rearNode.adag.getName());
                rearList.add(bigDax.adag.getName());
            }

        }

        return headNode;
    }
    protected void setSiteList(SiteAbstraction site, DAXAbstraction bigDax)
    {
        if(!mSiteList.containsKey(site))
            mSiteList.put(site, new ArrayList<DAXAbstraction>());
        ArrayList list = (ArrayList)mSiteList.get(site);
        if(!list.contains(bigDax))
            list.add(bigDax);
    }

    protected double getSiteSize(SiteAbstraction site, DAXAbstraction bigDax)
    {
       if(!mSiteList.containsKey(site))
            mSiteList.put(site, new ArrayList<DAXAbstraction>());
        ArrayList list = (ArrayList)mSiteList.get(site);
        double size = 0.0;
        HashMap entireMap = new HashMap<String, Double>();
        for(Iterator it = list.iterator(); it.hasNext();)
        {
            DAXAbstraction newDax = (DAXAbstraction)it.next();
            HashMap map = (HashMap)mDaxMap.get(newDax);
            //HashMap newMap = (HashMap)mDaxMap.get(bigDax);
            for(Iterator iit = map.keySet().iterator();iit.hasNext();)
            {
                String filename = (String)iit.next();
                double filesize = (Double)map.get(filename);
                if(!entireMap.containsKey(filename))
                {
                    entireMap.put(filename, filesize);
                    size += filesize;
                }
            }

        }

        if(!mDaxMap.containsKey(bigDax))
            mDaxMap.put(bigDax, new HashMap());
        HashMap newMap = (HashMap)mDaxMap.get(bigDax);
        HashMap localMap = new HashMap<String, Double>();
        double sumSize = 0.0;
        for(Iterator it = newMap.keySet().iterator();it.hasNext();)
        {
            String filename = (String)it.next();
            double filesize = (Double)newMap.get(filename);
            if(!entireMap.containsKey(filename))
            {
                entireMap.put(filename, filesize);
                size += filesize;
            }
            if(!mSitePureSize.containsKey(bigDax)&&!localMap.containsKey(filename))
            {
                localMap.put(filename, filesize);
                sumSize += filesize;
                
            }
        }
        if(!mSitePureSize.containsKey(bigDax))
        {
            mSitePureSize.put(bigDax, sumSize);
        }
        return size;
    }
    public void schedule()
    {
        System.err.println("Error should be implemented by its child");
    }

    @Override
    public void siteSelector()
    {
        getRel();

        String className = this.mBag.getPegasusProperties().getProperty("pegasus.subworkflow.estimator");
        mDaxQueue.clear();
        System.out.println("Selected Estimator is " + className);
        ArrayList estimatorList = new ArrayList<estimatorThread>();
        for(Iterator it = mDaxMap.keySet().iterator(); it.hasNext();)
        {
            DAXAbstraction bigDax =((DAXAbstraction)it.next());
            mDAXID2DAX.put(bigDax.adag.getName(), bigDax);
            if(!weight.containsKey(bigDax))
                weight.put(bigDax, new HashMap<SiteAbstraction, Double>());
            Map map = (Map)weight.get(bigDax);

            estimatorList.add(new estimatorThread(bigDax, null, mBag, mJob2Time,
                    className, mSiteIndex, map));

        }
        for(Iterator it = estimatorList.iterator(); it.hasNext(); )
        {
            estimatorThread et = (estimatorThread)it.next();
            et.getDone();
        }
        schedule();




    }



 



}
